package com.yelink.metrics.core;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.yelink.metrics.core.domain.TargetModel;

import cn.hutool.json.JSONUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;


/**
 * 消息监听
 * @author idea
 */
@Slf4j
@Component
public class MetricsConsumerListener {

    @Resource
    private TriggerHandler triggerHandler;

    @KafkaListener(topicPattern = "#{metricsTopic.getTriggerTopic()}")
    public void forwardMessage(List<ConsumerRecord<?, String>> consumerRecords) {
        log.info("【metrics消息】本次批量处理kafka消息数：{}", consumerRecords.size());
        for (ConsumerRecord<?, String> record : consumerRecords) {
            TargetModel targetModel = JSONUtil.toBean(record.value(), TargetModel.class);
            triggerHandler.trigger(targetModel);
        }
    }


}
